package net.bwie.jtp.dws.log.funcion;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.nimbusds.jwt.util.DateUtils;
import net.bwie.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;


public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {
    // TODO: 2025/5/19 声明变量
    private transient ValueState<String> lastVisitDateState = null;

    // TODO: 2025/5/19 定义一个地区字典数据方法
    private transient HashMap<String,String> areaDic  = null;

    // TODO: 2025/5/19 初始化变量
    @Override
    public void open(Configuration parameters) throws Exception {
//        lastVisitDateState = getRuntimeContext().getState(
//                new ValueStateDescriptor<String>("last-visit-date", String.class));
        // TODO: 2025/5/20 状态描述
        ValueStateDescriptor<String> lastVisitDateStates =
                new ValueStateDescriptor<>("lastVisitDateState", String.class);
        areaDic=getAreaDic();
        // TODO: 2025/5/20 优化ttl.设置状态的生命周期为1天
        StateTtlConfig build = StateTtlConfig
                .newBuilder(Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();
        lastVisitDateStates.enableTimeToLive(build);
        // TODO: 2025/5/20 初始化
        lastVisitDateState = getRuntimeContext().getState(lastVisitDateStates);

    }

    private HashMap<String, String> getAreaDic() {
        // TODO: 2025/5/19 map集合 
        HashMap<String, String> map = new HashMap<>();
        // TODO: 2025/5/19   添加数据
        map.put("110000","北京");
        map.put("120000","天津");
        map.put("140000","山西");
        map.put("150000","内蒙古");
        map.put("130000","河北");
        map.put("310000","上海");
        map.put("320000","江苏");
        map.put("330000","浙江");
        map.put("340000","安徽");
        map.put("350000","福建");
        map.put("360000","江西");
        map.put("370000","山东");
        map.put("710000","台湾");
        map.put("230000","黑龙江");
        map.put("220000","吉林");
        map.put("210000","辽宁");
        map.put("610000","陕西");
        map.put("620000","甘肃");
        map.put("630000","青海");
        map.put("640000","宁夏");
        map.put("650000","新疆");
        map.put("410000","河南");
        map.put("420000","湖北");
        map.put("430000","湖南");
        map.put("440000","广东");
        map.put("450000","广西");
        map.put("460000","海南");
        map.put("810000","香港");
        map.put("820000","澳门");
        map.put("510000","四川");
        map.put("520000","贵州");
        map.put("530000","云南");
        map.put("500000","重庆");
        map.put("540000","西藏");
        // TODO: 2025/5/19   返回集合
        return map ;
        
    }

    @Override
    public PageViewBean map(String value) throws Exception {
        // TODO: 2025/5/19 解析json数据
        JSONObject jsonObject = JSON.parseObject(value);
        // TODO: 2025/5/19 获取common
        JSONObject commonStream = jsonObject.getJSONObject("common");
        // TODO: 2025/5/19 获取page
        JSONObject pageObject = jsonObject.getJSONObject("page");
        // TODO: 2025/5/19 依据地区编码获取省份
        String arValue = commonStream.getString("ar");
        String province = areaDic.getOrDefault(arValue,arValue);
        // TODO: 2025/5/19 uv 访客数
        Long uvCount = 0L;
        // TODO: 2025/5/19 获取上一次访问日期
        String lastVisitDate = lastVisitDateState.value();
        // TODO: 2025/5/19 获取当前日期
        String visitDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);
        // TODO: 2025/5/19 值为null时表示今日第一次访问uv字段为1
        // TODO: 2025/5/19 值不为null时,上一次的访问日期不等于当前的日期,也表示今日第一次访问uv字段为1
        if(null== lastVisitDate || !lastVisitDate.equals(visitDate)){
            uvCount = 1L;
            // 更新日期状态
            lastVisitDateState.update(visitDate);
        }

        // TODO: 2025/5/19 返回实体类
        return  new PageViewBean(
                //开始时间
                null,
                //结束时间
                null,
                // 品牌
                commonStream.getString("ba"),
                // 渠道
                commonStream.getString("ch"),
                // 省份
                province,
                // 是否新用户
                commonStream.getString("is_new"),
                // pv 访问量
                pageObject.get("last_page_id") == null ? 1L : 0L,
                // session_count
                1L,
                // 访问时长
                pageObject.getLong("during_time"),
                // uvcount
                uvCount,
                // ts
                jsonObject.getLong("ts")
        );
    }
}
